package org.example.cdc;

import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.data.Struct;

public class CustomDeserializationSchema implements DebeziumDeserializationSchema<String> {
    /**
     * {
     * "db": "",
     * "tablename": "",
     * "before"：{"id":"","name":""},
     * ”after“: {"id":"","name":""},
     * "op":""
     * }
     *
     * @param record
     * @param out
     * @throws Exception
     */
    @Override
    public void deserialize(SourceRecord record, Collector<String> out) throws Exception {
        JSONObject result = new JSONObject();
        String topic = record.topic();
        String[] split = topic.split("\\.");
        String database = split[1];
        String table = split[2];
        result.put("db", database);
        result.put("tablename", table);
        JSONObject before = new JSONObject();
        Struct value = (Struct) record.value();
        if (value.schema().field("before") != null) {
            Struct beforeValue = (Struct) value.get("before");
            if (beforeValue != null) {
                for (Field field : beforeValue.schema().fields()) {
                    before.put(field.name(), beforeValue.get(field));
                }
            }
        }
        result.put("before", before);

        JSONObject after = new JSONObject();
        if (value.schema().field("after") != null) {
            Struct afterValue = (Struct) value.get("after");
            if (afterValue != null) {
                for (Field field : afterValue.schema().fields()) {
                    after.put(field.name(), afterValue.get(field));
                }
            }
        }
        result.put("after", after);
        //获取作类型
        Envelope.Operation operation =null;// Envelope.operationFor(record);
        result.put("op", operation);
        out.collect(result.toJSONString());
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }
}
